package com.xxx.debeziumdemo.config;

import io.debezium.connector.mysql.MySqlConnector;
import io.debezium.relational.history.FileDatabaseHistory;
import lombok.Data;
import org.apache.kafka.connect.storage.FileOffsetBackingStore;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.io.File;
import java.io.IOException;

/**
 * debezium配置文件
 * https://mp.weixin.qq.com/s/isfAyAiQCkn25gXgow6xqA
 */
@Configuration
@ConfigurationProperties(prefix = "debezium.datasource")
@Data
public class MysqlBinlogConfig {

    private String hostname;
    private String port;
    private String user;
    private String password;
    private String tableWhitelist;
    private String storageFile;
    private String historyFile;
    private Long flushInterval;
    private String serverId;
    private String serverName;

    @Bean(name = "MysqlBinlogConfig")
    public io.debezium.config.Configuration MysqlBinlogConfig() throws Exception {
        checkFile();
        io.debezium.config.Configuration configuration = io.debezium.config.Configuration.create()
                .with("name", "mysql_connector")
                .with("connector.class", MySqlConnector.class)                  // 监控的数据库类型，这里选mysql
                .with("offset.storage", FileOffsetBackingStore.class)           // 把读取进度存到本地文件
                // .with("offset.storage", KafkaOffsetBackingStore.class)       // 当使用kafka时，选KafkaOffsetBackingStore
                .with("offset.storage.file.filename", storageFile)              // 存放读取进度的本地文件地址
                .with("offset.flush.interval.ms", flushInterval)                // 读取进度刷新保存频率，默认1分钟
                .with("database.history", FileDatabaseHistory.class.getName())  //
                .with("database.history.file.filename", historyFile)            //
                .with("snapshot.mode", "Schema_only")                           // 快照模式
                .with("database.server.id", serverId)                           // 伪装成slave的Debezium服务的id,多个Debezium的话不能重复
                .with("database.server.name", serverName)                       // 伪装成slave的Debezium服务的name，多个Debezium的话不能重复
                .with("database.serverTimezone", "Asia/Shanghai")               // mysql时区
                .with("database.hostname", hostname)                            // host、port、用户名、密码
                .with("database.port", port)
                .with("database.user", user)
                .with("database.password", password)
//                .with("database.dbname", dbname)
                .with("table.whitelist", tableWhitelist)                        // 监控的表名白名单,建议设置此值，只监控这些表的binlog
//                .with("database.whitelist", "test")                           // 监控的数据库白名单,如果选此值，会忽略table.whitelist，然后监控此db下所有表的binlog
                .build();
        return configuration;

    }

    private void checkFile() throws IOException {
        String dir = storageFile.substring(0, storageFile.lastIndexOf("/"));
        File dirFile = new File(dir);
        if (!dirFile.exists()) {
            dirFile.mkdirs();
        }
        File file = new File(storageFile);
        if (!file.exists()) {
            file.createNewFile();
        }
    }
}
